﻿using DotNetty.Framework.ExtendChannelOfClient;
using DotNetty.Transport.Channels;
using DotNetty.Transport.Channels.Groups;
using iTool.ClusterComponent;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace DotNetty.Framework
{
    /// <summary>
    /// 为Channel绑定消息组
    /// </summary>
    public static class ExtendClientGroup
    {
        internal static ConcurrentDictionary<string, KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup>> dictionaryGroup;

        internal static ConcurrentDictionary<string, object> _lockDic;

        static ExtendClientGroup()
        {
            dictionaryGroup = new ConcurrentDictionary<string, KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup>>();
            _lockDic = new ConcurrentDictionary<string, object>();
        }

        /// <summary>
        /// 获取订阅用户组
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="group">用户组</param>
        /// <returns>是否获存在</returns>
        public static bool GetGroupByChannel(this IChannelHandlerContext context, string channel, out IChannelGroup group)
        {
            var isSuccess = dictionaryGroup.TryGetValue(channel, out KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup> keyValue);
            group = keyValue.Value;
            return isSuccess;
        }

        /// <summary>
        /// 订阅渠道消息
        /// </summary>
        /// <param name="context"></param>
        /// <param name="channel">订阅渠道名称</param>
        public static void Subscribe(this IChannelHandlerContext context, string channel)
        {
            var _lock = _lockDic.GetOrAdd(channel, (key) => new object());

            Func<string, KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup>> func = (a) =>
            {
                if (dictionaryGroup.ContainsKey(channel))
                    return default;

                var group = new DefaultChannelGroup(channel, context.Executor);
                group.Add(context.Channel);
                var observer = new ReceiveChannelNotifyObserver(channel, group);
                observer.StartAsync();
                return new KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup>(observer, group);
            };

            lock (_lock)
            {
                dictionaryGroup.AddOrUpdate(channel, func.Invoke(channel), (key, keyValue) =>
                {
                    // 说明被注销了
                    if (keyValue.Value == null)
                    {
                        return func.Invoke(channel);
                    }
                    else if (!keyValue.Value.Contains(context.Channel))
                    {
                        keyValue.Value.Add(context.Channel);
                        //bindObserver.Invoke();
                    }
                    return keyValue;
                });
            }
        }

        /// <summary>
        /// 订阅渠道消息，并通知该渠道一条消息
        /// </summary>
        /// <param name="context"></param>
        /// <param name="channel">订阅渠道名称</param>
        /// <param name="notifyMessage">通知的消息</param>
        public static async Task SubscribeAsync(this IChannelHandlerContext context, string channel, string notifyMessage)
        {
            // 订阅触发通知的通知消息
            {
                context.GetGroupByChannel(channel, out IChannelGroup group);
                if (group != null)
                {
                    await group.WriteAndFlushAsyncOfTextAsync(notifyMessage);
                }
            }

            context.Subscribe(channel);
        }

        /// <summary>
        /// 卸载指定订阅
        /// </summary>
        /// <param name="channel">渠道</param>
        /// <param name="context"></param>
        public static void UnSubscribe(this IChannelHandlerContext context, string topic)
        {
            UnSubscribe(context.Channel, topic);
        }

        public static void UnSubscribe(this IChannel iChannel, string topic)
        {
            if (dictionaryGroup.TryGetValue(topic, out KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup> keyValue))
            {
                keyValue.Value?.Remove(iChannel);
                if (keyValue.Value.Count < 1)
                {
                    dictionaryGroup.TryRemove(topic, out keyValue);
                    (keyValue.Key as ReceiveChannelNotifyObserver).UnsubscribeAsync().ConfigureAwait(false);
                }
            }
        }

        /// <summary>
        /// 卸载指定订阅，并通知该渠道一条消息
        /// </summary>
        /// <param name="channel"渠道</param>
        /// <param name="notifyMessage">通知消息</param>
        public static async Task UnSubscribeAsync(this IChannelHandlerContext context, string channel, string notifyMessage)
        {

            context.UnSubscribe(channel);

            // 订阅触发通知的通知消息
            {
                context.GetGroupByChannel(channel, out IChannelGroup group);
                if (group?.Count > 0)
                {
                    await group.WriteAndFlushAsyncOfTextAsync(notifyMessage);
                }
            }
        }

        /// <summary>
        /// 卸载通道订阅的通道
        /// </summary>
        /// <param name="context"></param>
        public static void UnSubscribeByChannel(this IChannelHandlerContext context)
        {
            UnSubscribeByChannel(context.Channel);
        }





        /// <summary>
        /// 卸载的一些考虑
        /// 我们是否不需要卸载？ 或者定期卸载。  否则可能会导致频繁的   订阅卸载
        /// </summary>
        /// <param name="channel"></param>

        public static void UnSubscribeByChannel(this IChannel channel)
        {
            // 使用Group锁
            var channelKey = channel.GetChannel();
            var _lock = _lockDic.GetOrAdd(channelKey, (key) => new object());

            var containsGroup = dictionaryGroup.Where(item => item.Value.Value == null || item.Value.Value?.Contains(channel) == true);
            if (containsGroup.Count() > 0)
            {
                foreach (var item in containsGroup)
                {
                    if (item.Value.Value == null)
                    {
                        lock (_lock)
                        {
                            if (dictionaryGroup.TryRemove(item.Key, out KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup> keyValue) && keyValue.Value.Count == 0)
                            {
                                (keyValue.Key as ReceiveChannelNotifyObserver).UnsubscribeAsync().ConfigureAwait(false);
                            }
                        }
                    }
                    else
                    {
                        lock (_lock)
                        {
                            item.Value.Value.Remove(channel);
                            if (item.Value.Value?.Count < 1)
                            {
                                if (dictionaryGroup.TryRemove(item.Key, out KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup> keyValue) && keyValue.Value.Count == 0)
                                {
                                    (keyValue.Key as ReceiveChannelNotifyObserver).UnsubscribeAsync().ConfigureAwait(false);
                                }
                            }
                        }
                    }
                }
            }
            else
            {
                lock (_lock)
                {
                    if (dictionaryGroup.TryGetValue(channelKey, out KeyValuePair<SubscribeQueueHandler<string>, IChannelGroup> keyValue) && keyValue.Value.Count == 0)
                    {
                        dictionaryGroup.TryRemove(channelKey, out _);
                        (keyValue.Key as ReceiveChannelNotifyObserver).UnsubscribeAsync().ConfigureAwait(false);
                    }
                }
            }


        }

    }
}
